################# 代码注解 begin #####################
"""


"""
################# 代码注解 end #####################

################# 项目经典片段 begin #####################

"""
一  -----------------
 创建消息    接口:/messages    
 涉及文件:api/message.py   
        config/config.py  
        utils/resp.py   

get------

    # 如果通过公众号消息返回所有文本消息和该公众号下的媒体消息
    messages = Message.query.filter(
        or_(Message.wx_app_id==wx_app_id,
            Message.wx_app_id==0),
        Message.deleted_at.is_(None)).order_by(desc(Message.id)).all()

     messages = Message.query.filter(Message.deleted_at.is_(None)).order_by(desc(Message.id)).all()


二-----------------------
创建一个用户群组    接口:/groups   
涉及文件:api/group.py,  
        daemon/task.py

class GroupListApi(Resource):

post---------

     parser.add_argument('wx_apps_id', type=list, location='json', required=True)
    logger.info(group.__dict__)

     update_total_count.delay(total_count_sql, group.id)
     logger.info('发送异步任务, 更新total_count')

    update_total_count.delay(query_sql, group_child.id)   异步任务

    task.py-----  异步任务都写在task.py
            query_sql = query_sql.replace(';', '')

        sql = 'select count(1) from ({}) as tmp'.format(query_sql)

        total_count = db.session.execute(sql).scalar()      ???

    if not group:
        # 应对生产环境的主从延迟  ???
        time.sleep(0.5)
    group = Group.query.get(group_id)  # 根据主键获取

    parent_group = Group.query.with_for_update(of=Group.total_count).get(group.parent_id)   ???



get----------

    父组子组, 创建群组时可以配置多个公众号的群组, 见test环境silo下的messages表的一个群组中四个公众号

     groups = Group.query.filter(Group.parent_id == 0).order_by(desc(Group.created_at)).all()

        sql = """select group_concat(`groups`.wx_app_id), `groups`.parent_id
                    as parent_id
                    from `groups`
                    join silo_wx_apps swa on `groups`.wx_app_id = swa.wx_app_id
                    where `groups`.parent_id != 0
                    group by `groups`.parent_id"""
        logger.info('\n获取子组wx_app_id列表的sql: \n{}'.format(sql))
        rows = db.session.execute(sql)

            wx_apps_id_dict[row.values()[1]] = list(map(int, row.values()[0].split(",")))


三-----------------------
   接口:/wx_apps   
   涉及文件:/wx_app.py    
            utils/query.py

class WxAppListApi(Resource):

post---------

        parser.add_argument('wx_apps', type=list, location='json', required=True)


get----------
    def get(self):
        """获取微信公众号列表接口"""

        logger.info('\n 收到获取微信公众号列表请求')
        apps = SiloWxApp.query.all()
        data = []
        for app in apps:
            dicts = {
                "app_name": app.app_name,
                "wx_app_id": app.wx_app_id,
                'app_id': app.app_id,
                'description': app.description,
            }
            data.append(dicts)
        return Resp.ok(data=data)



四-----------------------
任务    接口:/tasks   涉及文件:/task.py    utils/check.py  utils/query.py

post---------------
        # group_messages 格式为  [{1:3}, {5,9}]
        parser.add_argument('group_messages', type=list, required=True, location='json', default=[])

        # 如果不是任意数据检查都通过   
        if not all(map(lambda item:check_group_message(item), args.group_messages)):
            return Resp.bad_request(errmsg='group_messages数据有误', code=400)

        # 需要返回的数据
        data = dict()
        data.update(  # 为什么用update()
            title=task.title,
            task_id=task.id,
            description=task.description,
            executed_at=executed_at,
            created_at=str(task.created_at),
            updated_at=str(task.updated_at)
        )

  
    check.py---
      def check_group_message(group_message):
        if not isinstance(group_message, dict):
            return False
        if isinstance(group_message.get('msg_id'), int) and isinstance(group_message.get('group_id'), int):
            return True
        else:
            return False


        task = Task(title=args.title, description=args.description)
        db.session.add_all(group_messages+[task])
        db.session.commit()

       executed_at = str(task.executed_at) if task.executed_at else ''


get--------
    
    def get(self):
        """获取任务列表"""
        
        tasks = Task.query.order_by(desc(Task.created_at)).all()
        data = []
        for task in tasks:
            task_data = dict()
            task_data["task_id"] = task.id
            task_data["title"] = task.title
            task_data["description"] = task.description
            task_data["executed_at"] = str(task.executed_at) if task.executed_at else ''
            task_data["created_at"] = str(task.created_at)
            task_data["updated_at"]  = str(task.updated_at)
            data.append(task_data)
        return Resp.ok(data=data)
    

五--------------
执行任务    接口:/tasks/execute  涉及文件:/task.py    daemon/task.py   utils/query.py  utils/mock.py

        task = Task.query.get(args.task_id)

        res = db.session.query(GroupMessage.msg_id, TaskHasGroupMessage.id, Group.query_sql)\
            .join(TaskHasGroupMessage, TaskHasGroupMessage.group_msg_id == GroupMessage.id)\
            .join(Group, Group.id == GroupMessage.group_id)\
            .filter(TaskHasGroupMessage.task_id == task.id).all()


        # 将多个推送分发
        for msg_id, task_has_group_msg_id, query_sql in res:
            wx_app_forward.delay(msg_id, task_has_group_msg_id, query_sql, args.test_mode)

                    task.executed_at = datetime.now()

    query.py----
        first_row = res.fetchone()

    daemon/task.py----  写异步任务函数的地方
        raise NotImplementedError


    @app.task
    def ding_alarm(msg):
        """发送到钉钉报警群"""
        if Const.ENV in ('test', 'prod'):
            r = requests.post(DING_TALK_ALARM_URL, json={'msgtype': 'text', 'text': {'content': msg}})
            logger.info(f'status_code: {r.status_code}, content: {r.content}')

    data = mock_wechat_callback(wx_msg_id=mass_log.wx_msg_id, totalcount=len(openid_list))

    res = requests.post(url=Const.MASS_TEXT_URL, json=data)


六--------------
获取单个任务信息  
    
class TaskApi(Resource):

    def get(self, task_id):
        """获取单个任务信息"""

        task = Task.query.get(task_id)
        if not task:
            return Resp.bad_request(errmsg='任务不存在')
        if task.executed_at:
            group_messages = task_info_after_execute(task_id)
        else:
            group_messages = task_info_before_execute(task_id)
        
        data = {
            'task_id': task_id,
            "group_messages": group_messages
        }
        return Resp.ok(data=data)



deamon------------------------
consumer.py------
class WechatFeedbackConsumer(BaseCustomReplyConsumerStep):
    
    @property
    def queue(self):
        return Queue(name='silo',
                     exchange=self.exchange,
                     routing_key='silo.daemon.backend.#')
    
    @with_application
    def callback(self, body, message):
        message.ack()
        logger.info(body)
        
        msg_id = body.get('MsgID')
        if not msg_id:
            logger.info('不是想要的消息')
            return

        logger.info('wx_msg_id: {}'.format(msg_id))
        
        MassLog.query.filter(MassLog.wx_msg_id == msg_id).update({
            MassLog.total_count: body.get('TotalCount'),
            MassLog.filter_count: body.get('FilterCount'),
            MassLog.send_count: body.get('SentCount'),
            MassLog.error_count: body.get('ErrorCount'),
            MassLog.state: body.get('Status'),
        })
        
        db.session.commit()
        logger.info('群发日志已经更新: {}'.format(body.get('MsgID')))


app.steps['consumer'].add(WechatFeedbackConsumer)



settings.py---------------

# 指定导入的任务模块
CELERY_IMPORTS = (
    'daemon.task',
    'daemon.consumer'
)

# 指定celery队列
CELERY_QUEUES = (
    Queue('silo-backend', routing_key='silo.task.#'),
)

异步任务-------------------------
# celery 消息调度设置
CELERY_ROUTES = {
    'daemon.task.ding_alarm': {
        'queue': 'silo-backend',
        'routing_key': 'silo.task.ding-alarm',
    },
    'daemon.task.wx_app_forward': {
        'queue': 'silo-backend',
        'routing_key': 'silo.task.wx_app_forward'
    },
    'daemon.task.wx_app_mass_text': {
        'queue': 'silo-backend',
        'routing_key': 'silo.task.wx_app_mass_text'
    },
    'daemon.task.update_total_count': {
        'queue': 'silo-backend',
        'routing_key': 'silo.task.update_total_count'
    }
}



task.py---------
@app.task
def ding_alarm(msg):
    """发送到钉钉报警群"""
    if Const.ENV in ('test', 'prod'):
        r = requests.post(DING_TALK_ALARM_URL, json={'msgtype': 'text', 'text': {'content': msg}})
        logger.info(f'status_code: {r.status_code}, content: {r.content}')


@app.task
@with_application
def wx_app_forward(msg_id, task_has_group_msg_id, query_sql, test_mode):
    """
    微信公众号群发任务
    :param msg_id: 要群发的消息id
    :param query_sql: 查询群组人数sql
    :param task_has_group_msg_id: 任务与推送对象与消息的关系表id
    """
    message = Message.query.get(msg_id)
    app_id, openid_list = get_openid_list(query_sql)
    if message.msg_type != 'text':
        raise NotImplementedError
    
    # 循环调用群发接口
    length = len(openid_list)
    for index in range(0, length, 10000):
        wx_app_mass_text.delay(message.content, openid_list[index:index+10000],
                               task_has_group_msg_id, app_id, test_mode)
    logger.info('任务群发人数: {}'.format(length))
    

@app.task
@with_application
def wx_app_mass_text(content, openid_list, task_has_group_msg_id, app_id, test_mode):
    logger.info('群发公众号: {}'.format(app_id))
    logger.info('群发人数: {}'.format(len(openid_list)))
    logger.info('群发消息: {}'.format(content))
    data = {
        "content": content,
        "app_id": app_id,
        "group_or_users": openid_list,
    }
    if test_mode:
        # 模拟模拟
        mass_log = MassLog(
            task_has_group_msg_id=task_has_group_msg_id,
            # wx_msg_id=0,
            state=' test send job submission success',
        )
        db.session.add(mass_log)
        db.session.flush()
        mass_log.wx_msg_id = mass_log.id
        db.session.commit()
        # 向mq抛出数据
        queue = Queue(name='silo', exchange=Exchange(name='micro-events', type='topic'),
                      routing_key='silo.daemon.backend.mass')
        data = mock_wechat_callback(wx_msg_id=mass_log.wx_msg_id, totalcount=len(openid_list))
        product_worker(queue=queue, data=data)
        return 'test_mode'

    res = requests.post(url=Const.MASS_TEXT_URL, json=data)
    try:
        r = res.json()
    except JSONDecodeError:
        logger.info(res.status_code)
        logger.info(res.status_code)
        return
    
    logger.info(r)
    mass_log = MassLog(
        task_has_group_msg_id=task_has_group_msg_id,
        wx_msg_id=r['msg_id'],
        state=r['errmsg']
    )
    mass_log.save()
    return r


@app.task
@with_application
def update_total_count(query_sql, group_id):
    """更新群组的 total count"""
    
    query_sql = query_sql.replace(';', '')
    sql = 'select count(1) from ({}) as tmp'.format(query_sql)
    logger.info('\n获取查询集的总数sql:\n{}\n'.format(sql))

    total_count = db.session.execute(sql).scalar()

    try:
        total_count = db.session.execute(sql).scalar()
    except Exception as e:
        db.session.rollback()

    group = Group.query.get(group_id)
    if not group:
        # 应对生产环境的主从延迟
        time.sleep(0.5)
    group = Group.query.get(group_id)
    if not group:
        return {'msg': '不存在的group_id: {}'.format(group_id)}
    
    # 存在父组的情况
    if group.parent_id != 0:
        parent_group = Group.query.with_for_update(of=Group.total_count).get(group.parent_id)
        parent_group.total_count += total_count
        logger.info('父组 {} total_count加上: {}'.format(parent_group.title, total_count))
    logger.info('更新非父组 {} total_count为{}'.format(group.title, total_count))
    group.total_count = total_count
    db.session.commit()


"""
################# 项目经典片段 end #####################

################### 自己测试 begin #######################

"""
if not all(map(lambda item:check_group_message(item), args.group_messages)):


data = dict()
data.update(
    title=task.title,
    task_id=task.id,
    description=task.description,
    executed_at=executed_at,
    created_at=str(task.created_at),
    updated_at=str(task.updated_at)
)

"""
################### 自己测试 end #######################
